package com.hugedata.idc.etl;

import java.io.DataInputStream;
import java.io.IOException;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import com.hugedata.idc.utils.HDFSUtils;

public class QosSendHDFS
{

    public static void main(String[] args) throws IOException
    {
        CreateQosSendFiles(args[0], args[1]);

    }

    /**
     * @param srcDir
     * @param destDir
     * @throws IOException
     */
    public static void CreateQosSendFiles(String srcDir, String destDir) throws IOException
    {
        Configuration config = HDFSUtils.getConf();

        FileSystem fs = FileSystem.get(config);
        Path srcPath = new Path(srcDir);

        FileStatus[] licationsStatus = fs.listStatus(srcPath);

        for (int i = 0; i < licationsStatus.length; i++)
        {

            FileStatus location = licationsStatus[i];

            String locationname = location.getPath().getName();

            String loc = locationname.substring(10);
            if (location.isDirectory())
            {
                FileStatus[] isps = fs.listStatus(location.getPath());

                for (int j = 0; j < isps.length; j++)
                {
                    FileStatus isp = isps[j];

                    String ispname = isp.getPath().getName();

                    String lp = ispname.substring(5);

                    if (isp.isDirectory())
                    {

                        String filename = loc + "-" + lp + ".txt";

                        FileStatus[] files = fs.listStatus(isp.getPath());

                        for (int k = 0; k < files.length; k++)
                        {

                            if (files[k].isFile())
                            {
                                Path destPath = new Path(destDir, filename);

                                DataInputStream inputStream = fs.open(files[k].getPath());

                                FSDataOutputStream outputStream;
                                if (fs.exists(destPath))
                                {
                                    outputStream = fs.append(destPath);
                                }
                                else
                                {
                                    outputStream = fs.create(destPath);
                                }
                                IOUtils.copyLarge(inputStream, outputStream);
                                IOUtils.closeQuietly(inputStream);
                                IOUtils.closeQuietly(outputStream);
                            }
                        }
                    }
                }
            }
        }

    }
}
